package com.weic.spark.scala.p2

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @Auther:BigData-weic
 * @ClassName:_03CombineByKeyOps
 * @Date:2020/12/7 21:49
 * @功能描述: $FunctionDescription
 * @Version:1.0
 */
object _03CombineByKeyOps {
	def main(args: Array[String]): Unit = {
		val conf = new SparkConf()
    		.setAppName("_03CombineByKeyOps")
    		.setMaster("local[*]")
		val sc = new SparkContext(conf)

//		cbnOps(sc)
		cbnOps1(sc)


		sc.stop()
	}


	//模拟groupByKey
	def cbnOps1(sc: SparkContext)={
		case class Student(id: Int, name:String, province: String)
		val stuRDD = sc.parallelize(List(
			Student(1, "卫超", "山西"),
			Student(2, "昊天", "福建"),
			Student(3, "苏江", "上海"),
			Student(4, "大志", "山西"),
			Student(5, "龙哥", "山西"),
			Student(6, "浩东", "北京"),
			Student(7, "忠忠", "北京")
		),2).mapPartitionsWithIndex((index,partition) => {
			val parList = partition.toList
			println(s"分区编号：${index}-->分区:${parList.mkString("[",",","]")}")
			parList.toIterator
		})

		val proStuRDD: RDD[(String,Student)] = stuRDD.map(stu => (stu.province,stu))
		def createCombiner1(stu:Student): Array[Student]={
			Array[Student](stu)
		}
		//局部聚合
		def mergeValue1(arr:Array[Student],stu:Student): Array[Student]={
			arr.:+(stu)
		}
		def mergeCombiner1(arr:Array[Student],arr1:Array[Student]):Array[Student]={
			arr ++ arr1
		}
		val province2Infos:RDD[(String,Array[Student])] = proStuRDD.combineByKey(createCombiner1, mergeValue1, mergeCombiner1)

		province2Infos.foreach{case (province, stus) => {
			println(s"province2Infos>>> 省份：${province}, 学生信息：${stus.mkString(", ")}, 人数：${stus.size}")
		}}
	}

	//模拟reduce
	def cbnOps(sc: SparkContext)={
		val str = sc.parallelize(Array(
			"hello you",
			"hello you",
			"hello her",
			"hi his",
			"hello me"
		))
		val word = str.flatMap(words => words.split("\\s+")).map(word => (word,1))
		def createCombiner(num:Int):Int={
			num
		}
		def mergeValue(num:Int,sumi:Int):Int ={
			num + sumi
		}
		def mergeCombiner(sumi:Int,sum:Int):Int ={
			sum + sumi
		}
		val wordCount = word.combineByKey(createCombiner,mergeValue,mergeCombiner)
		wordCount.foreach{case(word,count) => {
			println(s"$word ----> $count")
		}}
	}
}
